package com.zte.claa.inficombo.csa.app.service.rptparse;

import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import com.zte.claa.inficombo.csa.app.config.CSAConfigProps;
import com.zte.claa.inficombo.csa.app.config.LocOptConfProps;
import com.zte.claa.inficombo.csa.app.config.RedisChannelProps;
import com.zte.claa.inficombo.csa.app.dao.mysql.impl.DevCmdConfDataMapper;
import com.zte.claa.inficombo.csa.app.dao.mysql.impl.LocLoRaBeaconMapper;
import com.zte.claa.inficombo.csa.app.model.dev.DevRunState;
import com.zte.claa.inficombo.csa.app.model.dev.DevTpExtData;
import com.zte.claa.inficombo.csa.app.service.correction.LocCorrectionService;
import com.zte.claa.inficombo.csa.app.service.csa.CSAMainService;
import com.zte.claa.inficombo.csa.app.service.csa.impl.BaseCacheManager;
import com.zte.claa.inficombo.csa.app.service.custparam.LocCustParamModelService;
import com.zte.claa.inficombo.csa.app.service.dataredeal.MetedataRedealService;
import com.zte.claa.inficombo.csa.app.service.dev.CSADevtypeService;
import com.zte.claa.inficombo.csa.app.service.devpower.DevPowerDataService;
import com.zte.claa.inficombo.csa.app.service.jsplugin.CSAJSPluginService;
import com.zte.claa.inficombo.csa.app.service.persistent.CSADataPersistService;
import com.zte.claa.inficombo.csa.app.service.photo.PhotoDataDealService;
import com.zte.claa.inficombo.csa.app.service.redis.CSAMessagePublisher;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.ClaaDtuUpdataSubParser;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.ClaaLocUpdataSubParser;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.ExtRptDataParser;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.ThirdPartyDtuUpdataSubParser;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.ThirdPartyRawUpdataSubParser;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.UpDataSubParser;
import com.zte.claa.inficombo.csa.app.service.rptparse.ext.claa.IClaaDevConfDataSenderService;
import com.zte.claa.inficombo.csa.app.util.JsonUtils;

/**
 * 本容器类提供的属性查询方法，仅限于各个Parser类调用，否则会因初始化时机导致null对象问题。。。
 * 
 * @author 0283000115
 *
 */

@Service
public class ParserContainer implements InitializingBean, DisposableBean {
	// 分包结果过期时间：30分钟。
	// private final static long MULTIPKG_EXP_TIME = 30*60*1000L;
	// 日志记录
	private final Logger logger = LoggerFactory.getLogger(ParserContainer.class);
	
	// dev run state process queue.
	private static final int QUEUE_MAX_SIZE = 1000;
	
	// 单个sensor的多包超期时长（3分钟）
	public static final long SENSOR_PKG_EXPTIME = 180*1000L;

	@Autowired
	private StringRedisTemplate redisStrTemplate;
	
	@Autowired
	private RedisChannelProps redisChannelProps;
	
	@Autowired
	private CSAConfigProps csaConfigProps;
	
	@Autowired
	private BaseCacheManager cacheManager;
	
	@Autowired
	private CSADataPersistService persisService;
	
	@Autowired
	private CSADevtypeService devtpService;
	
	@Autowired
	private CSAJSPluginService jsPnpService;
	
	@Autowired
	private CSAMainService csaMainService;
	
	@Autowired
	private IClaaDevConfDataSenderService devConfDataSenderService;
	
	@Autowired
	private DevCmdConfDataMapper devCmdConfMapper;
	
	@Autowired
	private MetedataRedealService meteDataRedealService;
	
	@Autowired
	private LocLoRaBeaconMapper locLoRaBeaconMapper;
	
	// csa.updata.3rd.subparser.devtps=huace_gnss,huace_rain_dist
	@Value("${csa.updata.3rd.subparser.devtps}")
	private String need3rdSubparserDevtps = "";
	
	@Value("${csa.updata.3rd.subparser.raw.devtps}")
	private String need3rdParseRawTransDevtps = "";
	
	@Value("${csa.updata.multi.pkg.exp.minutes}")
	private int multiPkgExpMinutes = 30;
	
	// 定位原始数据处理优化.
	@Autowired
	private LocOptConfProps locOptConfigure;
	
	
	@Autowired
	private LocCustParamModelService locCustParamModelService;
	
	@Autowired
	private LocCorrectionService locCorrectionService;
	
	@Autowired
	private PhotoDataDealService photoDataDealService;
		
	@Autowired
	private DevPowerDataService devPowerDataService;
		
	/**
	 * parser解析器映射表
	 * 针对主动上报数据、设备状态上报、命令异步响应上报的解析器。
	 */
	private Map<String, IParser> parserMap = new HashMap<String, IParser>();
	
	/**
	 * 针对主动上报数据的分类子解析器，主要是应对一些特殊逻辑的数据解析（如自研定位数据解析、原始码流上报等情况）
	 */
	private Map<String, UpDataSubParser> updSubParserMap = new HashMap<String, UpDataSubParser>();
	
	/**
	 * 分包数据临时缓存映射表.
	 * key: deveui, value: MultiPackageData
	 */
	private Map<String, MultiPackageData> multiPkgMap = new ConcurrentHashMap<String, MultiPackageData>();
	
	/**
	 * 上报数据外部处理钩子映射表（仅针对部分涉及外部解析或原始码流上报的场景，如华测GNSS设备）.
	 * 注意：需要相应的外部处理服务实现类主动到该map注册！！
	 * key: devtypeid;
	 * val: ExtRptDataParser;
	 */	
	private Map<String, ExtRptDataParser> extDealearMap = new HashMap<String, ExtRptDataParser>();
	
	
	
	// redis消息发布器.-->终端上报测量Mete数据发布；
	private CSAMessagePublisher updataPublisher;
	// redis消息发布器.-->状态上报数据发布（包括appEUI对应的CSIF链路状态、dev的join状态，online/offline状态；
	private CSAMessagePublisher statePublisher;
	// redis消息发布器.-->日志上报数据发布；
	private CSAMessagePublisher logPublisher;
	// redis消息发布器.-->终端上报告警Alarm数据发布；
	private CSAMessagePublisher alarmPublisher;
	// redis消息发布器.-->自研定位功能上报的定位原始数据发布（暂时未使用）；
	private CSAMessagePublisher claaLocRawDataPublisher;
	// redis消息发布器.-->自研融合定位终端上报的GPS坐标 数据发布（在base侧专门实现入库和二次加工）；
	private CSAMessagePublisher claaLocGpsCoordPublisher;
	// redis消息发布器.-->自研GSP图像采集终端通知AIC模块识别请求
	private CSAMessagePublisher aicPhotoAnalyPublisher;
	
	
	// ackcmd执行线程池.
    private ExecutorService execSvr;
    // 分包数据过期清理线程池.
    private ScheduledExecutorService expDataCleanScheduler = null;
    // need3rdSubparserDevtps对应的set
    private Set<String> thirdSubparserDevtpsSet = new HashSet<String>();
    
    // 根据上报数据设置设备在线状态处理器...
    private Thread runStateDealear = null;
	private LinkedBlockingQueue<DevRunState> queue = null;
	private boolean isWorkOn = true;
	
	/**
	 * 初始化.
	 */
	@Override
	public void afterPropertiesSet() throws Exception {
		// this.execSvr = Executors.newFixedThreadPool(1);
		this.execSvr = new ThreadPoolExecutor(1, 1, 60, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(100000));		
		this.expDataCleanScheduler = Executors.newScheduledThreadPool(1);
		this.updataPublisher = new CSAMessagePublisher(this.redisStrTemplate, this.redisChannelProps.getUpdatarptchannel());
		this.meteDataRedealService.setCSAMessagePublisher(this.updataPublisher);
		this.statePublisher = new CSAMessagePublisher(this.redisStrTemplate, this.redisChannelProps.getStaterptchannel());
		this.logPublisher = new CSAMessagePublisher(this.redisStrTemplate, this.redisChannelProps.getLogrptchannel());
		this.alarmPublisher = new CSAMessagePublisher(this.redisStrTemplate, this.redisChannelProps.getAlarmrptchannel());
		this.claaLocRawDataPublisher = new CSAMessagePublisher(this.redisStrTemplate, this.redisChannelProps.getClaalocrawdatarpt());
		this.claaLocGpsCoordPublisher = new CSAMessagePublisher(this.redisStrTemplate, this.redisChannelProps.getClaaLoccoorrpt());
		this.aicPhotoAnalyPublisher = new CSAMessagePublisher(this.redisStrTemplate, this.redisChannelProps.getAicPhotoAnaly());
		// 1.终端数据上报.
		ParserUpdata updataParser = new ParserUpdata();
		updataParser.setParserContainer(this);
		this.parserMap.put(updataParser.getType(), updataParser);
		
		// 2.CS强制下线通知.
		ParserCSForcedQuitData csForceQuitParser = new ParserCSForcedQuitData();
		csForceQuitParser.setParserContainer(this);
		this.parserMap.put(csForceQuitParser.getType(), csForceQuitParser);
		
		// 3.dev join通知.
		ParserDevJoinData devJoinParser = new ParserDevJoinData();
		devJoinParser.setParserContainer(this);
		this.parserMap.put(devJoinParser.getType(), devJoinParser);
		
		// 4.dev online/offline通知。
		ParserDevOnlineData devOnlineParser = new ParserDevOnlineData();
		devOnlineParser.setParserContainer(this);
		this.parserMap.put(devOnlineParser.getType(), devOnlineParser);
		// 兼容dev_online旧版本命令字.
		this.parserMap.put(ParserDevOnlineData.MSG_TYPE_V125, devOnlineParser);
		
		// 5.1 updata CLAA-loc-subparser 注册.
		ClaaLocUpdataSubParser claaLocSubParser = new ClaaLocUpdataSubParser();
		claaLocSubParser.setParserContainer(this);
		this.updSubParserMap.put(claaLocSubParser.getParserType(), claaLocSubParser);

		// 5.2 updata CLAA-dtu-subparser 注册.
		ClaaDtuUpdataSubParser claaDTUSubParser = new ClaaDtuUpdataSubParser();
		claaDTUSubParser.setParserContainer(this);
		this.updSubParserMap.put(claaDTUSubParser.getParserType(), claaDTUSubParser);
		
		// 5.3 updata 3rd-RAW-subparser 注册.
		ThirdPartyRawUpdataSubParser thirdPSubParser = new ThirdPartyRawUpdataSubParser();
		thirdPSubParser.setParserContainer(this);
		this.updSubParserMap.put(thirdPSubParser.getParserType(), thirdPSubParser);
		
		// 5.4  updata 3rd-DTU-subparser 注册.		
		ThirdPartyDtuUpdataSubParser thirdDtuSubParser = new ThirdPartyDtuUpdataSubParser();
		thirdDtuSubParser.setParserContainer(this);
		this.updSubParserMap.put(thirdDtuSubParser.getParserType(), thirdDtuSubParser);

		//分包数据过期清理.
		Runnable task = new Runnable(){
			@Override
			public void run() {
				cleanExpMultiPkgData();
			}
		};
		// 每隔1分钟调度一次.
		this.expDataCleanScheduler.scheduleAtFixedRate(task, 10, 60, TimeUnit.SECONDS);
		logger.info("[**]Expire data clean task is submitted and will be scheduled in 10 seconds later!");
		
		// need3rdSubparserDevtps
		this.thirdSubparserDevtpsSet.clear();
		if(need3rdSubparserDevtps != null && !need3rdSubparserDevtps.trim().equals("")){
			String[] dtpArr = this.need3rdSubparserDevtps.trim().split(",");
			for(String dtp : dtpArr){
				this.thirdSubparserDevtpsSet.add(dtp.trim());
			}
		}
		// 无特殊处理要求的原始码流透传类型配置.
		if(need3rdParseRawTransDevtps != null && !need3rdParseRawTransDevtps.trim().equals("")){
			String[] dtpArr = this.need3rdParseRawTransDevtps.trim().split(",");
			for(String dtp : dtpArr){
				this.thirdSubparserDevtpsSet.add(dtp.trim());
			}
		}
		
		// dev runstate process...
		this.isWorkOn = true;
		this.queue = new LinkedBlockingQueue<DevRunState>(QUEUE_MAX_SIZE);		
		this.runStateDealear = new Thread(new Runnable() {
			@Override
			public void run() {
				dealDevRunState();
			}
		});
		this.runStateDealear.setName("DevrunStateDealear-dispatcher");
		this.runStateDealear.start();	
	}
	
	@Override
	public void destroy() throws Exception {
		if(this.execSvr != null){
			this.execSvr.shutdown();
			this.execSvr = null;
		}
		this.isWorkOn = false;
		if(this.queue != null){
			this.queue.clear();
			this.queue.offer(new DevRunState());
		}
		
		// 过期数据清理定时器.
		if(this.expDataCleanScheduler != null){
			this.expDataCleanScheduler.shutdown();
			this.expDataCleanScheduler = null;
		}
		// 关闭缓存.
		this.cacheManager.shutdown();
		this.cacheManager = null;
	}

	/**
	 * 查询parser
	 * 
	 * @param msgType
	 * @return
	 */
	public IParser getParser(String msgType){
		return this.parserMap.get(msgType);
	}
	
	/**
	 * 子解析器获取
	 * 
	 * @param devType
	 * @return
	 */
	public UpDataSubParser getUpdSubParser(String devTypeId){
		if(devTypeId == null || devTypeId.trim().equals("")){
			return null;
		}
		// 0.第三方原始码流上传（如HC-GNSS&Rain&dist设备: huace_gnss, huace_rain_dist）.
		if(this.thirdSubparserDevtpsSet.contains(devTypeId))
		{
			return this.updSubParserMap.get(ThirdPartyRawUpdataSubParser.PARSER_TP_3RDP_RAW);
		}
		// 不需要根据devtype的fntype来判断的，都需要放上边.
		DevTpExtData dtpExt = this.cacheManager.getDevTpExtData(devTypeId);
		if(dtpExt == null){
			// 使用默认值.
			return null;
		}
		// 1.判断是否为自研定位设备: 4=自研纯GPS定位； 8=自研蓝牙定位锚点设备; 16=自研蓝牙定位终端设备; 32=Lora定位终端设备; 64=Lora定位信标设备;
		int fnType = dtpExt.getFntype();
		if((fnType & DevTpExtData.FN_TP_4)  != 0 || 
		   (fnType & DevTpExtData.FN_TP_8)  != 0 || 
		   (fnType & DevTpExtData.FN_TP_16) != 0 || 
		   (fnType & DevTpExtData.FN_TP_32) != 0 || 
		   (fnType & DevTpExtData.FN_TP_64) != 0 ||
		   (fnType & DevTpExtData.FN_TP_128) != 0) 
		{
			return this.updSubParserMap.get(ClaaLocUpdataSubParser.PARSER_TP_CLAALOC);
		}
		// 2.自研GSP/SSP及挂接的传感设备.
		else if((fnType & DevTpExtData.FN_TP_256) != 0 ||
		        (fnType & DevTpExtData.FN_TP_512) != 0) 
		{
			return this.updSubParserMap.get(ClaaDtuUpdataSubParser.PARSER_TP_CLAADTU);
		}
		// 第三方DTU及挂接的传感设备.
		else if((fnType & DevTpExtData.FN_TP_2048) != 0) 
		{
			return this.updSubParserMap.get(ThirdPartyDtuUpdataSubParser.PARSER_TP_3RDDTU);
		}		
		// 4.兼容极个别的根据设备类型实现的子解析器，目前暂时没有。
		return this.updSubParserMap.get(devTypeId);
	}
	
	/**
	 * 
	 * @param devtypeId
	 * @return
	 */
	public ExtRptDataParser getExtRptDataDealer(String devtypeId){
		return this.extDealearMap.get(devtypeId);
	}
	
	public void setExtRptDataDealer(String devtypeId, ExtRptDataParser dealer) {
		this.extDealearMap.put(devtypeId, dealer);
	}
	
	/**
	 * 查询分包数据.
	 * 
	 * @param deveui
	 * @return
	 */
	public MultiPackageData getMultiPkgData(String deveui) {
		return this.multiPkgMap.get(deveui);
	}
	
	/**
	 * 设置分包数据.
	 * 
	 * @param deveui
	 * @return
	 */
	public void setMultiPkgData(MultiPackageData data) {
		if(data == null){
			return;
		}
		this.multiPkgMap.put(data.getDeveui(), data);
	}
	
	/**
	 * 合并两个数据包对象.
	 * 考虑到多线程并发执行解析，这里需要做互斥（deveui级互斥）。
	 * 
	 * @param dest
	 * @return
	 */
	public MultiPackageData mergePackageData(MultiPackageData dest) {
		if (dest == null) {
			return null;
		}
		
		// 加锁逻辑统一放到parser中的调用方法中...
		// String devEui = dest.getDeveui();
		// synchronized (devEui) {  
		
		MultiPackageData curMpd = this.multiPkgMap.get(dest.getDeveui());
		if (curMpd == null) {
			this.multiPkgMap.put(dest.getDeveui(), dest);
			return dest;
		}
		if (dest == curMpd) {
			return curMpd;
		}
		curMpd.appendAllData(dest.getDataMap());
		if (dest.getIsConfirmed() == MultiPackageData.CONFIRMED_YES) {
			curMpd.setIsConfirmed(MultiPackageData.CONFIRMED_YES);
		}
		if (dest.getIsParsed() == MultiPackageData.PARSED_YES) {
			curMpd.setIsParsed(MultiPackageData.PARSED_YES);
		}
		if(curMpd.getMsgid() == MultiPackageData.INVALID_MSGID) {
			curMpd.setMsgid(dest.getMsgid());
		}
		// GSP多sensor交叉多包场景的特殊处理...
		// 1.交叉多包mode标记...
		if(curMpd.isGspCrossMpkMode() || dest.isGspCrossMpkMode()){
			curMpd.setMpModeOfGspCross();
		}
		// 2.GspSubCrossMPKData的merge...
		if(curMpd.isGspCrossMpkMode()){
			List<String> keyList = dest.getAllGspSubCrossMpkMapKeys();
			if(keyList != null && keyList.size() > 0){
				for(String key : keyList){
					GspSubCrossMPKData dstCrossData = dest.getGspSubCrossMPKData(key);
					if(dstCrossData == null || dstCrossData.getDatamap() == null){
						continue;
					}
					if(dstCrossData.isDataExpire(SENSOR_PKG_EXPTIME)){
						continue;
					}
					GspSubCrossMPKData curCrossData = curMpd.getGspSubCrossMPKData(key);
					if(curCrossData == null || curCrossData.isDataExpire(SENSOR_PKG_EXPTIME)){
						curMpd.setGspSubCrossMPKData(key, dstCrossData);
						continue;
					}
					Map<Integer, byte[]> mp = curCrossData.getDatamap();
					if(mp == null){
						curCrossData.setDatamap(dstCrossData.getDatamap());
					} else {
						mp.putAll(dstCrossData.getDatamap());
					}
					if(curCrossData.getRptTime() > dstCrossData.getRptTime()){
						curCrossData.setRptTime(dstCrossData.getRptTime());
					}
					if(dstCrossData.getTotalNum() > curCrossData.getTotalNum()){
						curCrossData.setTotalNum(dstCrossData.getTotalNum());
					}
				}
			}
		}
		return curMpd;
		
		// }
	}
	
	/**
	 * 删除分包数据.
	 * @param deveui
	 */
	public void removeMultiPkgData(String deveui) {
		this.multiPkgMap.remove(deveui);
	}
	
	/**
	 * 本容器类提供的属性查询方法，仅限于各个Parser类调用，否则会因初始化时机导致null对象问题。。。下同~~
	 * 
	 * @return
	 */
	public StringRedisTemplate getRedisStrTemplate() {
		return redisStrTemplate;
	}

	public RedisChannelProps getRedisChannelProps() {
		return redisChannelProps;
	}

	public BaseCacheManager getCacheManager() {
		return cacheManager;
	}

	public CSADataPersistService getPersisService() {
		return persisService;
	}

	public CSADevtypeService getDevtpService() {
		return devtpService;
	}

	public CSAMessagePublisher getUpdataPublisher() {
		return updataPublisher;
	}

	public CSAMessagePublisher getStatePublisher() {
		return statePublisher;
	}

	public CSAMessagePublisher getLogPublisher() {
		return logPublisher;
	}

	public CSAMessagePublisher getAlarmPublisher() {
		return alarmPublisher;
	}

	public CSAMessagePublisher getClaaLocRawDataPublisher() {
		return claaLocRawDataPublisher;
	}

	public CSAMessagePublisher getClaaLocGpsCoordPublisher() {
		return claaLocGpsCoordPublisher;
	}
	
	public CSAMessagePublisher getAicPhotoAnalyPublisher() {
		return aicPhotoAnalyPublisher;
	}

	public CSAJSPluginService getJsPnpService() {
		return jsPnpService;
	}

	public void setJsPnpService(CSAJSPluginService jsPnpService) {
		this.jsPnpService = jsPnpService;
	}

	public CSAConfigProps getCsaConfigProps() {
		return csaConfigProps;
	}
	
	public LocOptConfProps getLocOptConfigure() {
		return locOptConfigure;
	}

	public IClaaDevConfDataSenderService getDevConfDataSenderService() {
		return devConfDataSenderService;
	}

	public void setDevConfDataSenderService(IClaaDevConfDataSenderService devConfDataSenderService) {
		this.devConfDataSenderService = devConfDataSenderService;
	}

	public DevCmdConfDataMapper getDevCmdConfMapper() {
		return devCmdConfMapper;
	}

	public void setDevCmdConfMapper(DevCmdConfDataMapper devCmdConfMapper) {
		this.devCmdConfMapper = devCmdConfMapper;
	}

	public int getMultiPkgExpMinutes() {
		return multiPkgExpMinutes;
	}

	public void setMultiPkgExpMinutes(int multiPkgExpMinutes) {
		this.multiPkgExpMinutes = multiPkgExpMinutes;
	}

	public String getNeed3rdParseRawTransDevtps() {
		return need3rdParseRawTransDevtps;
	}

	public void setNeed3rdParseRawTransDevtps(String need3rdParseRawTransDevtps) {
		this.need3rdParseRawTransDevtps = need3rdParseRawTransDevtps;
	}

	
	public MetedataRedealService getMeteDataRedealService() {
		return meteDataRedealService;
	}

	public void setMeteDataRedealService(MetedataRedealService meteDataRedealService) {
		this.meteDataRedealService = meteDataRedealService;
	}

	/**
	 * 下发设备控制命令。
	 * 
	 * 主要用于设备上报数据要求立即回复响应的情况.
	 * 
	 * @param cmdwrapper
	 */
	public void sendDevCtrlCmd(AppCmdWrapper cmdwrapper){
		this.execSvr.execute(new Runnable(){
			@Override
			public void run() {
				try {
					String appEui = cmdwrapper.getAppeui();
					String devEui = cmdwrapper.getDeveui();
					boolean confirm = cmdwrapper.isConfirm();
					int port = cmdwrapper.getPort();
					String payload = JsonUtils.toJson(cmdwrapper.getPayload());
					csaMainService.SendDevCtrlCmdIgnoreRsp(appEui, devEui, confirm, port, payload, null);
				}catch(Exception e){
					logger.error("sendDevCtrlCmd failed: deveui="+cmdwrapper.getDeveui()+", err="+e.getMessage());
				}
			}
		});
	}
	
	/**
	 * 分包过期数据检查和清理.
	 */
	public void cleanExpMultiPkgData() {
		if(this.multiPkgMap.size() == 0){
			return;
		}
		try{
			Set<String> devEuiSet = new HashSet<String>();
			devEuiSet.addAll(this.multiPkgMap.keySet());
			long curTimes = System.currentTimeMillis();
			for(String deveui : devEuiSet){
				MultiPackageData mpd = this.multiPkgMap.get(deveui);
				if(mpd == null){
					continue;
				}
				if(curTimes - mpd.getRptTime() >= this.multiPkgExpMinutes*60*1000L){
					this.multiPkgMap.remove(deveui);
					logger.warn("Remove MPD of expire time["+this.multiPkgExpMinutes+"Min] for deveui: "+deveui);
				}
			}
		}catch(Exception ignore){}
	}
	
	// 添加到linkedqueue中，等待解析处理...
	public void addDevRunStateData(DevRunState state){
		boolean b = this.queue.offer(state);
		if(!b){
			logger.warn("[*!!*]Add dev runstate data to queue failed: "+state.toString());
		}
	}
	
	// 分发devrunstatedata
	private void dealDevRunState(){
		while (this.isWorkOn) {
			try {
				// 发送原始码流到Dest服务器进行数据转换...
				DevRunState stateData = this.queue.poll(120, TimeUnit.SECONDS);
				if(stateData == null){
					logger.debug("No dev run state data available now!");
					continue;
				} 
				if(stateData.getDevEUI() == null || stateData.getDevEUI().trim().equals("") ){
					continue;
				}
				this.persisService.updateDevRunState(stateData, false);				
				try{
					Thread.sleep(1);
				}catch(Throwable ignore){}
			} catch (Throwable t) {
				logger.error("dealDevRunState failed: " + t.getMessage(), t);
			}
		}
		logger.warn("Dealer thread exit: " + Thread.currentThread().getName());
	}

	public LocLoRaBeaconMapper getLocLoRaBeaconMapper() {
		return locLoRaBeaconMapper;
	}

	public void setLocLoRaBeaconMapper(LocLoRaBeaconMapper locLoRaBeaconMapper) {
		this.locLoRaBeaconMapper = locLoRaBeaconMapper;
	}

	public LocCustParamModelService getLocCustParamModelService() {
		return locCustParamModelService;
	}

	public LocCorrectionService getLocCorrectionService() {
		return locCorrectionService;
	}

	public PhotoDataDealService getPhotoDataDealService() {
		return photoDataDealService;
	}


	public DevPowerDataService getDevPowerDataService() {
		return devPowerDataService;
	}

	
}

